package org.example.analyzer;

import com.google.gson.Gson;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;
import org.example.hbase.HBaseUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.FileWriter;
import java.io.IOException;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.stream.Collectors;

public class DataAnalyzer {

    private static final Logger LOGGER = LoggerFactory.getLogger(DataAnalyzer.class);

    public String analyze() throws IOException {
        LOGGER.info("开始数据分析...");
        Map<String, Object> analysisResults = new HashMap<>();

        try (Table table = HBaseUtil.getConnection().getTable(HBaseUtil.TABLE_NAME)) {
            Scan scan = new Scan();
            scan.addFamily(HBaseUtil.CF_VEHICLE);
            scan.addFamily(HBaseUtil.CF_RECALL);

            LOGGER.info("正在从HBase表 '{}' 扫描数据...", HBaseUtil.TABLE_NAME);
            try (ResultScanner scanner = table.getScanner(scan)) {
                Map<String, Integer> recallsPerManufacturer = new HashMap<>();
                Map<String, Integer> recallsByYear = new HashMap<>();
                Map<String, Integer> recallsByComponent = new HashMap<>();
                Map<String, Integer> recallsByType = new HashMap<>();
                Map<String, Integer> recallsByNotificationMonth = new HashMap<>();
                Map<String, Map<String, Integer>> manufacturerTypeCrosstab = new HashMap<>();
                Map<String, Integer> recallsByMake = new HashMap<>();
                Map<String, Integer> recallsByModel = new HashMap<>();
                Map<String, Integer> recallsByRecallManufacturer = new HashMap<>();
                long totalEstimatedUnits = 0;

                int rowCount = 0;
                for (Result result : scanner) {
                    rowCount++;
                    String manufacturer = Bytes.toString(result.getValue(HBaseUtil.CF_VEHICLE, Bytes.toBytes("Vehicle Manufacturer")));
                    String year = Bytes.toString(result.getValue(HBaseUtil.CF_VEHICLE, Bytes.toBytes("Model Year")));
                    String component = Bytes.toString(result.getValue(HBaseUtil.CF_RECALL, Bytes.toBytes("Recall Component")));
                    String recallType = Bytes.toString(result.getValue(HBaseUtil.CF_RECALL, Bytes.toBytes("Recall Type")));
                    String notificationDate = Bytes.toString(result.getValue(HBaseUtil.CF_RECALL, Bytes.toBytes("Recall Notification Date")));
                    String vehicleMake = Bytes.toString(result.getValue(HBaseUtil.CF_VEHICLE, Bytes.toBytes("Vehicle Make")));
                    String vehicleModel = Bytes.toString(result.getValue(HBaseUtil.CF_VEHICLE, Bytes.toBytes("Vehicle Model")));
                    String recallManufacturer = Bytes.toString(result.getValue(HBaseUtil.CF_RECALL, Bytes.toBytes("Recall Manufacturer")));
                    String estimatedUnitsStr = Bytes.toString(result.getValue(HBaseUtil.CF_RECALL, Bytes.toBytes("Estimated Units")));

                    // 分析1: 各制造商的召回总数
                    if (manufacturer != null && !manufacturer.isEmpty()) {
                        recallsPerManufacturer.put(manufacturer, recallsPerManufacturer.getOrDefault(manufacturer, 0) + 1);
                    }
                    // 分析2: 各年份的召回总数
                    if (year != null && !year.isEmpty()) {
                        recallsByYear.put(year, recallsByYear.getOrDefault(year, 0) + 1);
                    }
                    // 分析3: 各召回组件的总数
                    if (component != null && !component.isEmpty()) {
                        recallsByComponent.put(component, recallsByComponent.getOrDefault(component, 0) + 1);
                    }
                    // 分析4: 按召回类型统计
                    if (recallType != null && !recallType.isEmpty()) {
                        recallsByType.put(recallType, recallsByType.getOrDefault(recallType, 0) + 1);
                    }
                    // 分析5: 随时间变化的召回趋势 (按月份)
                    if (notificationDate != null && notificationDate.length() >= 6) {
                        String month = notificationDate.substring(0, 6); // YYYYMM
                        recallsByNotificationMonth.put(month, recallsByNotificationMonth.getOrDefault(month, 0) + 1);
                    }
                    // 分析6: 制造商与召回类型的交叉分析
                    if (manufacturer != null && !manufacturer.isEmpty() && recallType != null && !recallType.isEmpty()) {
                        manufacturerTypeCrosstab.computeIfAbsent(recallType, k -> new HashMap<>())
                                .merge(manufacturer, 1, Integer::sum);
                    }
                    // 分析7: 各车辆品牌的召回总数
                    if (vehicleMake != null && !vehicleMake.isEmpty()) {
                        recallsByMake.put(vehicleMake, recallsByMake.getOrDefault(vehicleMake, 0) + 1);
                    }
                     // 分析8: 热门召回车型
                    if (vehicleModel != null && !vehicleModel.isEmpty()) {
                        recallsByModel.put(vehicleModel, recallsByModel.getOrDefault(vehicleModel, 0) + 1);
                    }
                     // 分析9: 召回制造商统计
                    if (recallManufacturer != null && !recallManufacturer.isEmpty()) {
                        recallsByRecallManufacturer.put(recallManufacturer, recallsByRecallManufacturer.getOrDefault(recallManufacturer, 0) + 1);
                    }
                    // 分析10: 累计预估影响单元
                    if (estimatedUnitsStr != null && !estimatedUnitsStr.isEmpty()) {
                        try {
                            totalEstimatedUnits += Long.parseLong(estimatedUnitsStr);
                        } catch (NumberFormatException e) {
                            // 忽略无法解析的字段，如 "MFR"
                        }
                    }
                }
                LOGGER.info("数据扫描完成，共处理 {} 行。", rowCount);

                // 获取排名前5的召回组件
                LOGGER.info("正在计算排名前5的召回组件...");
                Map<String, Integer> top5Components = recallsByComponent.entrySet().stream()
                        .sorted(Map.Entry.<String, Integer>comparingByValue().reversed())
                        .limit(5)
                        .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue, (e1, e2) -> e1, LinkedHashMap::new));

                Map<String, Integer> top10Manufacturers = getTopN(recallsPerManufacturer, 10);
                Map<String, Integer> top10Makes = getTopN(recallsByMake, 10);
                Map<String, Integer> top10Models = getTopN(recallsByModel, 10);

                analysisResults.put("top10Manufacturers", top10Manufacturers);
                analysisResults.put("recallsByYear", recallsByYear);
                analysisResults.put("top5RecallComponents", top5Components);
                analysisResults.put("recallsByType", recallsByType);
                analysisResults.put("recallsByNotificationMonth", recallsByNotificationMonth);
                analysisResults.put("manufacturerTypeCrosstab", manufacturerTypeCrosstab);
                analysisResults.put("top10Makes", top10Makes);
                analysisResults.put("top10Models", top10Models);
                analysisResults.put("recallsByRecallManufacturer", recallsByRecallManufacturer);
                analysisResults.put("totalEstimatedUnits", totalEstimatedUnits);
            }
        }

        // 将结果转换为JSON字符串并返回
        LOGGER.info("分析完成！正在将结果序列化为JSON...");
        return new Gson().toJson(analysisResults);
    }
    
    private <K> Map<K, Integer> getTopN(Map<K, Integer> map, int n) {
        return map.entrySet().stream()
                .sorted(Map.Entry.<K, Integer>comparingByValue().reversed())
                .limit(n)
                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue, (e1, e2) -> e1, LinkedHashMap::new));
    }
} 